#include "sdk.h"
#include "M_Code_threadpool.h"
#include <algorithm>
#include <functional>

M_Code_ThreadPool::~M_Code_ThreadPool() {
  wxMutexLocker lock( m_Mutex );
  std::for_each( m_threads.begin(), m_threads.end(), std::mem_fun( &cbWorkerThread::Abort ) );
  Broadcast();
  std::for_each( m_tasksQueue.begin(), m_tasksQueue.end(), std::mem_fun_ref( &cbThreadedTaskElement::Delete ) );
}

void M_Code_ThreadPool::SetConcurrentThreads( int concurrentThreads ) {
  if( concurrentThreads <= 0 ) {
    concurrentThreads = wxThread::GetCPUCount();
    if( concurrentThreads == -1 ) {
      m_concurrentThreads = 1;
    }
  }
  if( concurrentThreads == m_concurrentThreads ) {
    m_concurrentThreadsSchedule = 0;
    return;
  }
  wxMutexLocker lock( m_Mutex );
  _SetConcurrentThreads( concurrentThreads );
}

void M_Code_ThreadPool::_SetConcurrentThreads( int concurrentThreads ) {
  if( !m_workingThreads ) {
    std::for_each( m_threads.begin(), m_threads.end(), std::mem_fun( &cbWorkerThread::Abort ) );
    Broadcast();
    m_threads.clear();
    m_semaphore = CountedPtr<wxSemaphore>( new wxSemaphore( 0, concurrentThreads ) );
    m_concurrentThreads = concurrentThreads;
    m_concurrentThreadsSchedule = 0;
    for( std::size_t i = 0; i < static_cast<std::size_t>( m_concurrentThreads ); ++i ) {
      m_threads.push_back( new cbWorkerThread( this, m_semaphore ) );
      m_threads.back()->Create( m_stackSize );
      m_threads.back()->Run();
    }
  } else {
    m_concurrentThreadsSchedule = concurrentThreads;
  }
}

void M_Code_ThreadPool::AddTask( cbThreadedTask *task, bool autodelete ) {
  if( !task ) {
    return;
  }
  wxMutexLocker lock( m_Mutex );
  m_tasksQueue.push_back( cbThreadedTaskElement( task, autodelete ) );
  m_taskAdded = true;
  if( !m_batching && m_workingThreads < m_concurrentThreads ) {
    AwakeNeeded();
  }
}

void M_Code_ThreadPool::AbortAllTasks() {
  wxMutexLocker lock( m_Mutex );
  std::for_each( m_threads.begin(), m_threads.end(), std::mem_fun( &cbWorkerThread::AbortTask ) );
  std::for_each( m_tasksQueue.begin(), m_tasksQueue.end(), std::mem_fun_ref( &cbThreadedTaskElement::Delete ) );
  m_tasksQueue.clear();
}

void M_Code_ThreadPool::BatchEnd() {
  wxMutexLocker lock( m_Mutex );
  m_batching = false;
  AwakeNeeded();
}

M_Code_ThreadPool::cbThreadedTaskElement M_Code_ThreadPool::GetNextTask() {
  wxMutexLocker lock( m_Mutex );
  if( m_tasksQueue.empty() ) {
    return cbThreadedTaskElement();
  }
  cbThreadedTaskElement element = m_tasksQueue.front();
  m_tasksQueue.pop_front();
  return element;
}

void M_Code_ThreadPool::WorkingThread() {
  wxMutexLocker lock( m_Mutex );
  ++m_workingThreads;
}

bool M_Code_ThreadPool::WaitingThread() {
  wxMutexLocker lock( m_Mutex );
  --m_workingThreads;
  if( m_workingThreads <= 0 && m_tasksQueue.empty() ) {
    if( m_taskAdded ) {
      CodeBlocksEvent evt = CodeBlocksEvent( cbEVT_THREADTASK_ALLDONE, m_ID );
      wxPostEvent( m_pOwner, evt );
      m_taskAdded = false;
    }
    if( m_concurrentThreadsSchedule ) {
      _SetConcurrentThreads( m_concurrentThreadsSchedule );
      return false;
    }
  } else {
    m_semaphore->Post();
  }
  return true;
}

void M_Code_ThreadPool::TaskDone( M_Code_unused cbWorkerThread* thread ) {
  CodeBlocksEvent evt = CodeBlocksEvent( cbEVT_THREADTASK_ENDED, m_ID );
  wxPostEvent( m_pOwner, evt );
}

M_Code_ThreadPool::cbWorkerThread::cbWorkerThread( M_Code_ThreadPool *pool, CountedPtr<wxSemaphore> &semaphore )
  : m_abort( false ),
    m_pPool( pool ),
    m_semaphore( semaphore ),
    m_pTask( nullptr ) {
}

wxThread::ExitCode M_Code_ThreadPool::cbWorkerThread::Entry() {
  bool workingThread = false;
  while( !Aborted() ) {
    if( workingThread ) {
      workingThread = false;
      if( !m_pPool->WaitingThread() ) {
        break;
      }
      m_semaphore->Wait();
    }
    if( Aborted() ) {
      break;
    }
    if( !workingThread ) {
      m_pPool->WorkingThread();
      workingThread = true;
    }
    M_Code_ThreadPool::cbThreadedTaskElement element = m_pPool->GetNextTask();
    {
      wxMutexLocker lock( m_taskMutex );
      m_pTask = element.task;
    }
    if( !m_pTask ) {
      continue;
    }
    if( !Aborted() ) {
      m_pTask->Execute();
      {
        wxMutexLocker lock( m_taskMutex );
        m_pTask = nullptr;
        element.Delete();
      }
      m_pPool->TaskDone( this );
    }
  }
  if( workingThread ) {
    m_pPool->WaitingThread();
  }
  return nullptr;
}

void M_Code_ThreadPool::cbWorkerThread::Abort() {
  m_abort = true;
  AbortTask();
}

bool M_Code_ThreadPool::cbWorkerThread::Aborted() const {
  return m_abort;
}

void M_Code_ThreadPool::cbWorkerThread::AbortTask() {
  wxMutexLocker lock( m_taskMutex );
  if( m_pTask ) {
    m_pTask->Abort();
  }
}
